package com.lcy.util;


import com.lcy.bean.PageC;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

public class FlinkToHbase extends RichSinkFunction<PageC> {
    Connection conn;
    @Override
    public void open(Configuration parameters) throws Exception {
        org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "hadoop-single");
        conn = ConnectionFactory.createConnection(config);
    }
    @Override
    public void invoke(PageC value, Context context) throws Exception {
        Table table = conn.getTable(TableName.valueOf("kunkun"));
        Put put = new Put(Bytes.toBytes(value.getOs_name()));
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(value.toString()));
        table.put(put);
    }

    @Override
    public void close() throws Exception {
        conn.close();
    }
}
